package com.llbnk.springcoding.thread;


import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.ConfigService;
import com.ctrip.framework.apollo.model.ConfigChange;
import com.ctrip.framework.apollo.model.ConfigChangeEvent;
import com.google.common.collect.ImmutableSet;
import com.llbnk.springcoding.util.ThreadExcutorUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.xmlbeans.impl.xb.xsdschema.Public;

import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static com.llbnk.springcoding.thread.ThreadEnum.ThreadExcutorName.THREAD_POOL_NAME_PREFIX;

@Slf4j
public class ThreadExcutorServiceRefresher {

    /**
     * 静态代码块在类加载环节，被JVM执行目的是注册匿名加载器
     * ImmutableSet用于创建一个不可变的集合,使用apollo的模糊匹配
     */
    static {
        ConfigService.getAppConfig().addChangeListener(configChangeEvent -> {
            refreshThreadPool(configChangeEvent);
        },null, ImmutableSet.of(THREAD_POOL_NAME_PREFIX.getValue()));
    }



    public static void refreshThreadPool(ConfigChangeEvent configChangeEvent) {
        for(String key : configChangeEvent.changedKeys()){
            //1.获取apollo改动的线程池配置
            ConfigChange cfgChange = configChangeEvent.getChange(key);
            //add log and monitor
            log.info("thread pool start change programme,key:{},cfgChange:{}",key,cfgChange);
            //2.检查是否改造线程池存在
            String poolName = key.substring(THREAD_POOL_NAME_PREFIX.getValue().length());
            ThreadExcutorWrapper wrapper = ThreadExcutorFactory.getThreadExcutor(poolName);
            if(Objects.isNull(wrapper)){
                log.info("thread pool name not exist,key:{},cfgChange:{}",key,cfgChange);
            }
            //3.校验配置是否合法
            ThreadExcutorConfig newCfg = ThreadExcutorUtils.convert(cfgChange.getNewValue(),
                    (t) -> JSON.parseObject(t, ThreadExcutorConfig.class),
                    null);
            if(Objects.isNull(newCfg) || !newCfg.isValid()){
                log.warn("apollo config happen some mistake,newCfg:{},cfgChange:{}",newCfg,cfgChange);
            }
            //4.开始重新配置参数
            log.info("thread pool name start change now,newCfg:{},cfgChange:{}",newCfg,cfgChange);
            refreshExcutor(poolName,wrapper,newCfg);
        }

    }

    private static synchronized void refreshExcutor(String poolName, ThreadExcutorWrapper wrapper, ThreadExcutorConfig cfg) {
        try {
            ThreadPoolExecutor executor = (ThreadPoolExecutor) wrapper.getExecutorService();
            ChangeCapacityBlockingQueue<?> queue = (ChangeCapacityBlockingQueue<?>) executor.getQueue();
            //线程池队列长度不能缩小
            if(queue.size() > cfg.getCapacity()){
                log.warn("thread pool capacity not change small,poolName:{},cfg:{}",poolName,cfg);
                return ;
            }
            queue.setCapacity(cfg.getCapacity());
            executor.setCorePoolSize(cfg.getCore());
            executor.setKeepAliveTime(cfg.getKeepAliveTime(), TimeUnit.SECONDS);
            executor.setMaximumPoolSize(cfg.getMax());
            executor.allowCoreThreadTimeOut(cfg.isAllowCoreThreadTimeOut());

            if(queue.size() > 0){
                executor.prestartAllCoreThreads();
            }
            log.info("thread pool name:{} refresh success",poolName);
        } catch (Exception e) {
            log.error("thread pool name:{} refresh fail",poolName);
        } finally {
            ThreadExcutorUtils.refreshReportExcutor(poolName);
        }
    }

}

